package com.thering;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableReducer;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.Text;

import java.io.IOException;
import java.nio.charset.StandardCharsets;

/**
 * author: the_ring
 * date: 2022/6/8
 * function: movie info
 */
public class MovieInfoReducer extends TableReducer<Text, Text, ImmutableBytesWritable> {

    private byte[] date;
    private byte[] ratingFamily = "rating".getBytes(StandardCharsets.UTF_8);
    private byte[] ratingCountFamily = "rating_count".getBytes(StandardCharsets.UTF_8);
    private String[] info;
    private Put put;

    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        Configuration conf = context.getConfiguration();
        this.date = conf.get("date").getBytes(StandardCharsets.UTF_8);
    }


    @Override
    protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
        for (Text value : values) {
            info = value.toString().split(" ");

            // rating
            put = new Put(Bytes.toBytes(key.toString()));
            put.addColumn(ratingFamily, date, Bytes.toBytes(info[0]));
            context.write(null, put);

            // rating_count
            put = new Put(Bytes.toBytes(key.toString()));
            put.addColumn(ratingCountFamily, date, Bytes.toBytes(info[1]));
            context.write(null, put);

            System.err.println(key.toString() + " " + info[0] + " " + info[1]);
        }
    }
}
